package com.lagou.edu.message.server;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;

import com.corundumstudio.socketio.AckRequest;
import com.corundumstudio.socketio.AuthorizationListener;
import com.corundumstudio.socketio.Configuration;
import com.corundumstudio.socketio.HandshakeData;
import com.corundumstudio.socketio.SocketConfig;
import com.corundumstudio.socketio.SocketIOClient;
import com.corundumstudio.socketio.SocketIOServer;
import com.corundumstudio.socketio.listener.ConnectListener;
import com.corundumstudio.socketio.listener.DataListener;
import com.corundumstudio.socketio.listener.DisconnectListener;
import com.corundumstudio.socketio.listener.ExceptionListener;
import com.corundumstudio.socketio.namespace.Namespace;
import com.corundumstudio.socketio.protocol.Packet;
import com.corundumstudio.socketio.protocol.PacketType;
import com.corundumstudio.socketio.store.StoreFactory;
import com.lagou.edu.common.jwt.JwtUtil;
import com.lagou.edu.common.string.GsonUtil;
import com.lagou.edu.message.api.dto.Message;
import com.lagou.edu.message.consts.Constants;
import com.lagou.edu.message.server.store.StoreFacotryProvider;
import com.lagou.edu.message.util.PushUtils;

import com.lagou.edu.message.util.ServerConfigUtils;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

@Slf4j
public class PushServer {

    public static final PushServer pushServer = new PushServer();
    private Namespace pushNamespace;
    private SocketIOServer server;

    private PushServer() {

        final Configuration config = new Configuration();
        //Redisson
        config.setStoreFactory(StoreFacotryProvider.getRedissonStoreFactory());
        //Auth
        config.setAuthorizationListener(new UserAuthorizationListener());
        config.setPort(ServerConfigUtils.instance.getWebSocketPort());
        config.setContext(ServerConfigUtils.instance.getWebSocketContext());
        config.setOrigin(ServerConfigUtils.instance.getWebSocketOrigin());
        final SocketConfig socketConfig = new SocketConfig();
        //打开地址重用 ，netty中的内容
        socketConfig.setReuseAddress(true);
        config.setSocketConfig(socketConfig);
        server = new SocketIOServer(config);
        pushNamespace = (Namespace)server.addNamespace(ServerConfigUtils.instance.getWebSocketContext());
        config.setExceptionListener(new ExceptionListener() {
            @Override
            public void onEventException(Exception e, List<Object> list, SocketIOClient socketIOClient) {
                UUID sessionId = socketIOClient.getSessionId();
                log.error("onEventException sessionId:{},roomList:{}",sessionId, socketIOClient.get(Constants.ROOM),e);
            }

            @Override
            public void onDisconnectException(Exception e, SocketIOClient socketIOClient) {
                UUID sessionId = socketIOClient.getSessionId();
                log.error("onEventException sessionId:{},roomList:{}",sessionId, socketIOClient.get(Constants.ROOM),e);
            }

            @Override
            public void onConnectException(Exception e, SocketIOClient socketIOClient) {
                UUID sessionId = socketIOClient.getSessionId();
                log.error("onEventException sessionId:{},roomList:{}",sessionId, socketIOClient.get(Constants.ROOM),e);
            }

            @Override
            public void onPingException(Exception e, SocketIOClient socketIOClient) {
                UUID sessionId = socketIOClient.getSessionId();
                log.error("onEventException sessionId:{},roomList:{}",sessionId, socketIOClient.get(Constants.ROOM),e);
            }

            @Override
            public boolean exceptionCaught(ChannelHandlerContext channelHandlerContext, Throwable throwable) throws Exception {
                Channel channel = channelHandlerContext.channel();
                if(Objects.nonNull(channel)){
                    log.error("exception channel id:{}, isOpne:{}, remoteAddress:{}",channel.id(),channel.isOpen(),channel.remoteAddress());
                }
                return false;
            }
        });

        pushNamespace.addEventListener("send-message", Map.class, new DataListener<Map>() {
            @Override
            public void onData(SocketIOClient socketIOClient, Map data, AckRequest ackRequest) throws Exception {
                JwtUtil.JwtResult user = UserAuthorizationListener.getUserInfo(socketIOClient.getHandshakeData());
                Integer toUserId = user.getUserId();
                String content = (String)data.get("content");
                String type = (String)data.get("type");
                Message message = new Message(type, content, toUserId);
                log.info("from {} userid{} type {} content {}",user.getUserId(), user, type, content);
                push(message);
            }
        });
        pushNamespace.addDisconnectListener(new DisconnectListener() {
            @Override
            public void onDisconnect(SocketIOClient socketIOClient) {
                UUID sessionId = socketIOClient.getSessionId();
                JwtUtil.JwtResult user = UserAuthorizationListener.getUserInfo(socketIOClient.getHandshakeData());
                if(Objects.nonNull(user)){

                }else{
                    socketIOClient.disconnect();
                    return;
                }
                String userId = Integer.toString(user.getUserId());
                List<String> roomList = PushUtils.getRoomList(userId, null, null);
                roomList.forEach(socketIOClient::leaveRoom);
                try{
                    //离开房间
                    List<String> oldRoomList = socketIOClient.get(Constants.ROOM);
                    if(!CollectionUtils.isEmpty(oldRoomList)){
                        oldRoomList.forEach( s->{
                            if(!StringUtils.isEmpty(s)){
                                socketIOClient.leaveRoom(s);
                            }
                        });
                    }
                }catch (Exception e){
                    log.error("leave old room eception ,sid:{}",sessionId, e);
                }
                finally {
                    socketIOClient.del(Constants.ROOM);
                }
            }
        });
        pushNamespace.addConnectListener(new ConnectListener() {
            @Override
            public void onConnect(SocketIOClient socketIOClient) {
                UUID sessionId = socketIOClient.getSessionId();
                JwtUtil.JwtResult userInfo = UserAuthorizationListener.getUserInfo(socketIOClient.getHandshakeData());
                if(Objects.nonNull(userInfo)){
                    Set<String> allRooms = socketIOClient.getAllRooms();
                    allRooms.forEach(socketIOClient::leaveRoom);
                    String userId = Integer.toString(userInfo.getUserId());
                    List<String> roomList = PushUtils.getRoomList(userId, null,null);
                    roomList.forEach(socketIOClient::joinRoom);
                    socketIOClient.set(Constants.ROOM, roomList);
                }
            }
        });

    }

    /**
     * 推送消息
     *
     * @param message
     */
    public void push(Message message) {
    	final String type;
    	final Integer userId;
    	final String json;
    	long l11;
    	try{
    	    long l0 = System.currentTimeMillis();
    	    type = message.getType();
    	    userId = message.getUserId();
    	    json = GsonUtil.toJson(message);
            l11 = System.currentTimeMillis();
            if(l11 - 10 > 501 ){
                log.info("当前node 1-0 耗时:{}", l11 - l0);
            }
        }finally {
        }
    	String room;
    	long l12;
    	try{
    	    if(Objects.isNull(userId)){
    	        throw new NullPointerException("userId 不能为空");
            }
    	    room = PushUtils.getRoom(null, userId, null);
    	    l12 = System.currentTimeMillis();
    	    if(l12 -l11 >501){
                log.info("当前node 2-1 耗时:{}", l12 - l11);
            }

        }finally {
        }
        Packet packet;
    	long l13;
    	try{
            //组装对象
            packet = new Packet(PacketType.MESSAGE);
            packet.setSubType(PacketType.EVENT);
            packet.setName("message");
            ArrayList<Object> data = new ArrayList<>();
            data.add(json);
            packet.setData(data);
            packet.setNsp(pushNamespace.getName());

            l13 = System.currentTimeMillis();
            if(l13 -l12 >501){
                log.info("当前node 3-2 耗时:{}", l13 - l12);
            }
        }finally {
        }
    	AtomicInteger i1 = new AtomicInteger();
    	final long l2;
    	try{
    	    //当前服务推送
            i1.set(0);
            Iterable<SocketIOClient> roomClients = pushNamespace.getRoomClients(room);
            roomClients.forEach( s -> {
                s.send(packet);
                i1.getAndIncrement();
            });
            l2 = System.currentTimeMillis();
            if(l2 - l13 >501){
                log.info("当前node 2-13 耗时:{}", l2 - l13);
            }
        }finally {

        }

    }

    /**
     * 同步启动服务；
     */
    public void start() {
        try {
            server.start();
        } catch (Exception e) {
            log.error("Push server start failed!", e);
            System.exit(-1);
        }
    }

    /**
     * 停止服务
     */
    public void stop() {
        server.stop();
    }

    public Map<String, Object> getStatus() {
        HashMap<String, Object> status = new HashMap<>();
        status.put("namespace", pushNamespace.getName());   // namespace
        status.put("rooms", pushNamespace.getRooms());
        status.put("clients", pushNamespace.getAllClients().size());
        return status;
    }

    public void pushEveryOne(Message message) {
        Packet packet;

        try{
            //组装对象
            packet = new Packet(PacketType.MESSAGE);
            packet.setSubType(PacketType.EVENT);
            packet.setName("message");
            ArrayList<Object> data = new ArrayList<>();
            data.add(GsonUtil.toJson(message));
            packet.setData(data);
            packet.setNsp(pushNamespace.getName());

        }finally {
        }
        Set<String> rooms = pushNamespace.getRooms();
        rooms.forEach(room -> {
            Iterable<SocketIOClient> roomClients = pushNamespace.getRoomClients(room);
            roomClients.forEach( s -> {
                s.send(packet);
            });
        });
        log.info("pushEveryOne SUCCESS data:{}",GsonUtil.toJson(message));
    }
}
